package each_webcode

import common.HbaseTools
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable
import scala.collection.mutable.HashMap
import scala.io.Source
import scala.util.parsing.json.JSON;
object transferValueCode_to_Name_ilo {
        def main(args: Array[String]): Unit = {
                val sparkConf = new SparkConf().setAppName("ILOdatachange")//.setMaster("local[*]")
                val sc = new SparkContext(sparkConf)
                val baseInfotable: String = new String("BaseMapInfos") //DetailMapInfos
                val detailTable: String = new String("DetailMapInfos") //BaseMapInfos

                val inpath = args(0)
                val outpath = args(1)
                val numspartion = args(2)
                //val inpath = "D:\\InternationalData\\makeChange\\src\\main\\resources\\sub_ILO_Value.json"
                        //val outpath = "hdfs://node01:8020//tmp/InternationalData_tmp/sub_ilo_test_16"
                        //val numspartion =2
                        val webcodes = List("ILO")
                //make a cache:
                        val inputFile = inpath //读取json文件
                val jsonStr = sc.textFile(inputFile) //.sample(true,0.01);

                val result = jsonStr.map(s => JSON.parseFull(s))

                val rdds = result.map {
                        r =>
                                r match {
                                        case Some(maps: Map[String, Any]) => {
                                                val webcode = "ILO"
                                                val thashmap: HashMap[String, String] = new HashMap[String, String]()
                                                maps.foreach(t => {
                                                        thashmap += (t._1.toString -> t._2.toString.replace("\n",""))
                                                        })
                                                //maybe there should deal with _id
                                                thashmap("_id") = thashmap("_id").substring(12, 36)
                                                // get webcode + sourcecode
                                                // rowkeys
                                                //处理 sourcecode
                                                thashmap("SourceCode") = thashmap("SourceCode").replace("C://工作/ILO重做/ILOValue2\\", "")
                                                //tuple(随机数;ILO;sourcecode,map(json的key,json的value))
                                                (scala.util.Random.nextInt(80).toString + ";" + webcode + ";" + thashmap("SourceCode"), thashmap)
                                                //("0"+"_WTO"+"_"+thashmap("SourceCode"),thashmap)
                                                }
                                        }
                        }.groupByKey(numspartion.toInt).map {
                        case (k, v) => {
                                val k1array = k.split(";")
                                //ILO_SourceCode
                                        val k1 = k1array(1) + "_" + k1array(2)
                                //ILO_
                                        val pk1 = k1array(1) + "_"
                                //by k,you will get the mapinfos
                                        val tools = new HbaseTools()
                                val connection = tools.getConnection()

                                //map(kbase表的列名,列对应type类型)
                                        //表名:"BaseMapInfos",rowkey:ILO_,列族"dealTypes"-------------------------------------------找不到该代码
                                        //getFamily_column_value(connection:Connection,tableName: String, rowkey: String, columnFamily: String)
                                        val dealTypes: Map[String, String] = tools.getFamily_column_value(connection,baseInfotable, pk1, "dealTypes").toMap

                                //map(value表的key,kbase表的key)
                                        //表名"BaseMapInfos",rowkey:ILO_,列族"map01"-------------------------------------------找不到该代码
                                        val map01: Map[String, String] = tools.getFamily_column_value(connection,baseInfotable, pk1, "map01").toMap

                                //map(value表的列名,concept表的conceptId列值)
                                        //表名"BaseMapInfos",rowkey:ILO_,列族"map02"-------------------------------------------找不到该代码
                                        val map02: Map[String, String] = tools.getFamily_column_value(connection,baseInfotable, pk1, "map02").toMap
                                //create a map for map04
                                //then for each v;you can transfer code to value

                                //map("WebInfo",map(列名,列值))
                                        //map(表名"DetailMapInfos",rowkey:ILO_下的每一个列族,map(列名,列值))
                                        val map04: HashMap[String, Map[String, String]] = new HashMap[String, Map[String, String]]()
                                //getFamilys_list(connection:Connection,tableName: String, rowkey: String)
                                        //需要添加公用部分的数据，ILO_
                                        //publicfamilys:表名"DetailMapInfos",rowkey:ILO_下的所有列族
                                        val publicfamilys = tools.getFamilys_list(connection,detailTable,pk1)

                                //向map04中添加元素:(表名"DetailMapInfos",rowkey:ILO_下的每一个列族,map(列名,列值))------------------------------未知数据来源
                                for (onefamily: String <- publicfamilys) {
                                        //返回表名"DetailMapInfos",rowkey:ILO_,的每个列族下的map(列名,列值)
                                                //getFamily_column_value(connection:Connection,tableName: String, rowkey: String, columnFamily: String)
                                                val tmp = tools.getFamily_column_value(connection,detailTable, pk1, onefamily)
                                        map04.put(onefamily, tmp.toMap)
                                        }

                                //向map04中添加元素:("WebInfo",map(列名,列值))---来自makeBaseMap01_step01
                                        //get "DetailMapInfos","ILO_ZWE_M"
                                        //        WebInfo:SourceCode      timestamp=1610077994958, value=ZWE_M
                                        //        WebInfo:SourceName      timestamp=1610077994958, value=Zimbabwe_M
                                        //        WebInfo:WebCode         timestamp=1610077994958, value=ILO
                                        //        WebInfo:WebName         timestamp=1610077994958, value=\xE5\x9B\xBD\xE9\x99\x85\xE5\x8A\xB3\xE5\xB7\xA5\xE7\xBB\x84\xE7\xBB\x87
                                        //map04 的数据是独有的
                                        val tmp = tools.getFamily_column_value(connection,detailTable, k1, "WebInfo")
                                map04.put("WebInfo", tmp.toMap)

                                //special
                                        //得到表名为"BaseMapInfos",rowkey为"public",列族为"public_Time"的所有列名和列值组成的map(detailname,detailId)---------------------------来自代码make_Dic
                                        val PU_timeMap:Map[String,String] = tools.getFamily_column_value(connection,baseInfotable,"public","public_Time").toMap
                                //得到表名为"BaseMapInfos",rowkey为"public",列族为public_country"的所有列名和列值组成的map(detailname,detailId)-------------------------来自代码make_Dic
                                        val PU_country:Map[String,String] = tools.getFamily_column_value(connection,baseInfotable,"public","public_country").toMap
                                connection.close()

                                val onekeysValue = v.map {
                                        detailValuecode => { // get one record
                                                var oneLinceMap:HashMap[String,String] = new mutable.HashMap[String,String]()
                                                //map03
                                                        var map03 = detailValuecode
                                                dealTypes.keys.foreach {
                                                        keys =>
                                                                dealTypes(keys) match {

                                                                        case "type01" => {
                                                                                //map01 + map03
                                                                                if (map01.contains(keys) && map03.contains(map01(keys))) {
                                                                                        oneLinceMap.put(keys,map03(map01(keys)))
                                                                                        }
                                                                                }
                                                                                case "type02" => {
                                                                                        //map01 + map03 + map04
                                                                                        oneLinceMap.put(keys,map04("WebInfo")((map01(keys))))
                                                                                        }
                                                                                case "type03" => {
                                                                                        /*map01 + map02 =conceptid
                                                                                       map01 + map03 =detailId
                                                                                       conceptid + detailid + map04
                                                                                      */
                                                                                        if (map01.contains(keys) && map03.contains(map01(keys)) && map02.contains(map01(keys)) && map04.contains(map02(map01(keys))) && map04(map02(map01(keys))).contains(map03(map01(keys)))) {

                                                                                                oneLinceMap.put(keys, map04(map02(map01(keys)))(map03(map01(keys))))
                                                                                                }

                                                                                        }
                                                                                case "type04" => {
                                                                                        //map01 + map02 = conceptId
                                                                                        oneLinceMap.put(keys,map02(map01(keys)))
                                                                                        }
                                                                                case "type05" => {
                                                                                        //map01 + map03 = detailId
                                                                                        //one_column_value += ((keys+"\t"+map03(map01(keys)))+"(<>)")
                                                                                        if (map01.contains(keys) && map03.contains(map01(keys)) && map04.contains(map01(keys)) && map04(map01(keys)).contains(map03(map01(keys)))) {
                                                                                                oneLinceMap.put(keys,map04(map01(keys))(map03(map01(keys))))
                                                                                                }
                                                                                        }
                                                                                case "type06" => {
                                                                                        // map01 , a changeless value
                                                                                        if(map03.contains(map01(keys))){
                                                                                                oneLinceMap.put(keys,map01(keys))
                                                                                                }
                                                                                        }
                                                                                case "type08" => {
                                                                                        //map01 +  map02   get
                                                                                                // COUNTRY;DATASOURCE;DHSMICSGEOREGION;GBDREGION;MGHEREG;REGION;UNREGION;UNSDGREGION;WHOINCONMEREGION;WORLDBANKINCOMEGROUP
                                                                                                //+ map04 get dic
                                                                                                //map01 + map03 get code
                                                                                                val code = map03(map01(keys))
                                                                                        val dics = map02(map01(keys)).split(";")
                                                                                        var status = true
                                                                                        for (dickey <- dics) {
                                                                                                if (status && map04(dickey).contains(code)) {
                                                                                                        oneLinceMap.put(keys, map04(dickey)(code))
                                                                                                        status = false
                                                                                                        }
                                                                                                }
                                                                                        }
                                                                                case "type07" => {
                                                                                        // conceptid and detailid from differ column
                                                                                        //
                                                                                        if (map01.contains(keys) && map03.contains(map01(keys)) && map04.contains(map03(map01(keys)))) {
                                                                                                oneLinceMap.put(keys,map04(map03(map01(keys)))("ConceptName"))
                                                                                                }
                                                                                        }
                                                                                case "type09" => {
                                                                                        //map04(map01(k))("ConceptName")
                                                                                        if (map01.contains(keys) && map04.contains(map01(keys)) && map04(map01(keys)).contains("ConceptName")) {
                                                                                                oneLinceMap.put(keys, map04(map01(keys))("ConceptName"))
                                                                                                }
                                                                                        }
                                                                                case "type01_ilo" => {
                                                                                        // example:map01:(FootNote->note_indicator;note_classif)
                                                                                                //         map02:(note_indicator->note_indicator_en)
                                                                                                //         map02:(note_classif->note_classif_en)
                                                                                                var result: Array[String] = Array("", "")
                                                                                        val cells = map01(keys).split(";")
                                                                                        var i = 0
                                                                                        for (cell: String <- cells) {
                                                                                                //map03()
                                                                                                if (map03.contains(cell) && map02.contains(cell) && map04.contains(map02(cell)) && map04(map02(cell)).contains(map03(cell))) {
                                                                                                        result(i) = map04(map02(cell))(map03(cell))
                                                                                                        } else {

                                                                                                        }
                                                                                                i += 1
                                                                                                }
                                                                                        if(result.mkString("")!=""){
                                                                                                oneLinceMap.put(keys,result.mkString("，"))
                                                                                                }
                                                                                        }
                                                                        }
                                                        }
                                                //合并字段，生成
                                                //检测 Year字段，
                                                if(oneLinceMap.contains("Year")&&oneLinceMap("Year")!=""){
                                                        if(oneLinceMap("Year").length>4 && !oneLinceMap.contains("Period")){
                                                                oneLinceMap.put("Period",oneLinceMap("Year").substring(4))
                                                                oneLinceMap("Year") = oneLinceMap("Year").substring(0,4)
                                                                }
                                                        //检测Period字段
                                                        //1检查是否有值，
                                                        }
                                                //生成TimeSort字段；Period字段判断是否存在并进行转换，
                                                if(oneLinceMap.contains("Year")&&oneLinceMap.contains("Period")&&oneLinceMap("Period").trim!=""){
                                                        if(PU_timeMap.contains(oneLinceMap("Period").trim)){
                                                                oneLinceMap.put("TimeSort",oneLinceMap("Year")+PU_timeMap(oneLinceMap("Period").trim))
                                                                }else{
                                                                oneLinceMap.put("TimeSort",oneLinceMap.getOrElse("Year",""))
                                                                }
                                                        }else{
                                                        oneLinceMap.put("TimeSort",oneLinceMap.getOrElse("Year",""))
                                                        }
                                                //生成FuzzyIndicator字段
                                                if(oneLinceMap.contains("Indicator")&&oneLinceMap("Indicator")!="")
                                                        {oneLinceMap.put("FuzzyIndicator",oneLinceMap("Indicator"))
                                                                //生成IndicatorLength字段
                                                                oneLinceMap.put("IndicatorLength",oneLinceMap("Indicator").length.toString)
                                                                }
                                                //生成FuzzyReporter字段
                                                if(oneLinceMap.contains("Reporter")&&oneLinceMap("Reporter")!=""){
                                                        if(PU_country.contains(oneLinceMap("Reporter"))){
                                                                oneLinceMap.put("FuzzyReporter",oneLinceMap("Reporter")+"，"+PU_country(oneLinceMap("Reporter")))
                                                                }else{
                                                                oneLinceMap.put("FuzzyReporter",oneLinceMap("Reporter"))
                                                                }
                                                        }
                                                //Parterner
                                                //生成FUZZYPARTERNER字段
                                                if(oneLinceMap.contains("Parterner")&&oneLinceMap("Parterner")!=""){
                                                        if(PU_country.contains(oneLinceMap("Parterner"))){
                                                                oneLinceMap.put("FUZZYPARTERNER",oneLinceMap("Parterner")+"，"+PU_country(oneLinceMap("Parterner")))
                                                                }else{
                                                                oneLinceMap.put("FUZZYPARTERNER",oneLinceMap("Parterner"))
                                                                }
                                                        }
                                                val MetaData:Array[String] = Array("WebCode", "SourceCode", "ReporterCode", "Reporter", "ParternerCode", "Parterner", "Year", "Period",
                                                        "IndicatorCategory", "Indicator", "Dim1Type", "Dim1", "Dim2Type", "Dim2", "Dim3Type", "Dim3",
                                                        "Dim4Type", "Dim4", "Dim5Type", "Dim5", "Dim6Type", "Dim6", "Dim7Type", "Dim7", "Dim8Type", "Dim8" )
                                                val DIMTYPEMETADATA:Array[String] =Array("Dim1Type", "Dim2Type", "Dim3Type", "Dim4Type", "Dim5Type", "Dim6Type", "Dim7Type", "Dim8Type")
                                                val DIMMETADATA:Array[String] = Array("Dim1", "Dim2", "Dim3", "Dim4", "Dim5", "Dim6", "Dim7", "Dim8")

                                                //生成DIMTYPEMETADATA
                                                        var dimtypemetadata = DIMTYPEMETADATA.flatMap(oneLinceMap.getOrElse(_,"<->")+"，").mkString("").replace("<->，","")

                                                //生成DIMMETADATA
                                                        var dimemetadata = DIMMETADATA.flatMap(
                                                        onecel=>{
                                                                oneLinceMap.getOrElse(onecel,"<->")+"，"
                                                                }
                                                        ).mkString("").replace("<->，","")
                                                //生成MetaData
                                                        var metadata = MetaData.flatMap(oneLinceMap.getOrElse(_,"<->")+"，").mkString("").replace("<->，","")
                                                oneLinceMap.put("DIMTYPEMETADATA",dimtypemetadata)
                                                oneLinceMap.put("DIMMETADATA",dimemetadata)
                                                oneLinceMap.put("MetaData",metadata)
                                                //val kkl:RDD[String] = sc.parallelize(().toString)
                                                oneLinceMap.mkString("(<>)").replace(" -> ","\t")
                                                //val kkl:RDD[String] = sc.parallelize(().toString)
                                                }
                                        }
                                var fi = onekeysValue.mkString( "\n")
                                fi
                                }
                        }.saveAsTextFile(outpath)
                sc.stop()
                }

        }
